#include "SendMq.h"
#include "Exception.h"
#include <amqp.h>
#include "AppLication/CustomServerApplication.h"

/*====================================================
 * Name 	  			: SendMq
 * Function  			: SendMq
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/12/29
 * ==================================================*/
SendMq::SendMq()
{
    LOG.information("SendMq Start");
    channelid = 1;
    MQFlag = false;
    if (_cException == nullptr)
        _cException = new CException();
}
/*====================================================
 * Name 	  			: Amqp_Init_Mq
 * Function  			: Amqp_Init_Mq
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/12/29
 * ==================================================*/
void SendMq::Amqp_Init_Mq(const char* _rabbitmqIp,int _rabbitmqPort,const char* _rabbitmqUser,const char* _rabbitmqPasswd,
                          const char* _MessagePush_QueueName,
                          const char* _MessagePush_RoutKey,
                          const char* _rabbitmqStatus)
{
    try{
        int ret = 0;
        LOG.information("MQ Start Connection");
        ret = Amqp_New_Connection(_rabbitmqIp,_rabbitmqPort,_rabbitmqUser,_rabbitmqPasswd);
        if(ret < 0)
        {
            LOG.error("Amqp_New_Connection Failed:%s,%d",__FILE__,__LINE__);
            MQFlag = true;
            return;
        }
        ret = Amqp_Exchange_Declare("publish","topic",1,0);
        if(ret < 0)
        {
            LOG.error("Amqp_Exchange_Declare Failed:%s,%d",__FILE__,__LINE__);
            MQFlag = true;
            return;
        }
        //message status
        if(strcmp(_rabbitmqStatus, "ON") == 0)
        {
            ret = Amqp_Queue_Declare("publish","topic",_MessagePush_QueueName,_MessagePush_RoutKey,1,0);
            if(ret < 0)
            {
                LOG.error("Amqp_Queue_Declare Failed:%s,%d",__FILE__,__LINE__);
                MQFlag = true;
                return;
            }
            else
            {
                MQFlag = false;
                LOG.information("MQ Create Sucessful");
            }

        }
        else
        {
            LOG.information("MQ Function is Close");
            return;
        }
        LOG.information("MQ Connection Sucessful");

    }catch(Poco::Exception& e)
    {
        LOG.error("Amqp Init Falied:%s,%d",e.message(),__LINE__);
        MQFlag = true;
        return;
    }
}
/*====================================================
 * Name 	  			: ~SendMq
 * Function  			: ~SendMq
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/12/29
 *
 * ==================================================*/
SendMq::~SendMq()
{
    if(_cException != nullptr)
    {
        _cException->die_on_amqp_error(amqp_channel_close(conn,1,AMQP_REPLY_SUCCESS),"Closing Channel");;
        _cException->die_on_amqp_error(amqp_connection_close(conn,AMQP_REPLY_SUCCESS),"Closing connection");;
        _cException->die_on_error(amqp_destroy_connection(conn),"Ending Connection");
    }
    LOG.information("MQ Quit Sucessful");
    delete(_cException);
    _cException = nullptr;

}
int SendMq::Amqp_New_Connection(const char* _rabbitmqIp,int _rabbitmqPort,const char* _rabbitmqUser,const char* _rabbitmqPasswd)
{
    try{
        int sockfd;
        amqp_channel_open_ok_t * channelstr=nullptr;

//        std::cout << "===========" << CONF.getString("rabbitmqIp").c_str()<<
//                     CONF.getInt("rabbitmqPort") << std::endl;

        conn = amqp_new_connection();
        sockfd = amqp_open_socket(_rabbitmqIp, _rabbitmqPort);
        _cException->die_on_error(sockfd ,"Opening Socket");

        amqp_set_sockfd(conn, sockfd);
        _cException->die_on_amqp_error(amqp_login(conn,"/",0,131072, 0, AMQP_SASL_METHOD_PLAIN, _rabbitmqUser,_rabbitmqPasswd),"Logging in");
        channelstr = amqp_channel_open(conn, channelid);
        if(channelstr  == nullptr)
        {
            LOG.error("Amqp_New_Connection Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }

    }catch(Poco::Exception& e)
    {
        LOG.fatal("Amqp_New_Connection :%s,%d",e.message(),__LINE__);
        return -1;
    }
    return 0;

}
/*====================================================
 * Name 	  			: C_Amqp_Exchange_Declare
 * Function  			: Create exchange
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/07/22
 *
 * ==================================================*/
int SendMq::Amqp_Exchange_Declare(const char* _exchange, const char* _exchangeType, int _durable, int _autodelete)
{
    //创建exchange
    try
    {
        amqp_exchange_declare_ok_t * exchangedeclare=nullptr;
        exchangedeclare = amqp_exchange_declare(conn, channelid,amqp_cstring_bytes(_exchange),amqp_cstring_bytes(_exchangeType),0,_durable,_autodelete,0,amqp_empty_table);
        if(exchangedeclare  == nullptr)
        {
            LOG.error("amqp_exchange_declare Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }
    }
    catch (Poco::Exception& e)
    {
        LOG.fatal("Amqp_Exchange_Declare:%s,%d",e.message(),__LINE__);
        return -1;
    }
    return 0;
}

/*====================================================
 * Name 	  			: C_Amqp_Queue_Delete
 * Function  			: Queue Delete
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/08/2
 *
 * ==================================================*/
int SendMq::Amqp_Queue_Delete(const char* _queuename,string deviceid)
{
    try{
        char queuename[100];
        amqp_queue_delete_ok_t *deletequeue=NULL;
        sprintf(queuename,"%s%s",_queuename,deviceid.c_str());
        deletequeue = amqp_queue_delete(conn,channelid,amqp_cstring_bytes(queuename),1,1);
        if(deletequeue  == nullptr)
        {
            LOG.error("C_Amqp_Queue_Delete Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }

    }catch(Poco::Exception& e)
    {
        LOG.fatal("Amqp_Queue_Delete:%s,%d",e.message(),__LINE__);
        return -1;
    }
    return 0;
}

/*====================================================
 * Name 	  			: C_Amqp_Queue_Unbind
 * Function  			: Queue UnBind
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/07/25
 *
 * ==================================================*/
int SendMq::Amqp_Queue_Unbind(const char* _exchange,const char*_queuename,const char* _routingkey,string deviceid)
{
    try{
        amqp_queue_unbind_ok_t * ubindqueue=NULL;
        char queuename[100];
        sprintf(queuename,"%s%s",_queuename,deviceid.c_str());
        ubindqueue = amqp_queue_unbind(conn, channelid, amqp_cstring_bytes(queuename),amqp_cstring_bytes(_exchange),amqp_cstring_bytes(_routingkey),amqp_empty_table);
        if(ubindqueue  == nullptr)
        {
            LOG.error("amqp_queue_unbind Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }

    }catch(Poco::Exception& e)
    {
        LOG.fatal("Amqp_Queue_Unbind:%s,%d",e.message(),__LINE__);
        return -1;
    }
    return 0;
}

/*====================================================
 * Name 	  			: C_Amqp_Exchange_Declare
 * Function  			: Create exchange
 * InputParameter 	: void
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/07/25
 *
 * ==================================================*/
int SendMq::Amqp_Queue_Declare(const char* _exchange,const char* _exchangeType,const char* _queuename, const char* _routingkey,int _durable, int _autodelete)
{
    try{
        //创建queue
        amqp_queue_declare_ok_t *queuedeclare = NULL;
        amqp_queue_bind_ok_t * bindqueue = NULL;
        queuedeclare = amqp_queue_declare(conn, channelid, amqp_cstring_bytes(_queuename),0,_durable,0,_autodelete,amqp_empty_table);
        if(queuedeclare == nullptr)
        {
            LOG.error("create Queu Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }
        //bind exchange queue,其实这里是给客户端创建用的
        bindqueue = amqp_queue_bind(conn, channelid, amqp_cstring_bytes(_queuename), amqp_cstring_bytes(_exchange),amqp_cstring_bytes(_routingkey),amqp_empty_table);
        if(bindqueue == nullptr)
        {
            LOG.error("Bind Queu Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }
    }catch(Poco::Exception& e)
    {
        LOG.fatal("Amqp_Queue_Declare:%s,%d",e.message(),__LINE__);
        return -1;
    }
    return 0;
}

/*====================================================
 * Name 	  			: C_Amqp_First_Consume
 * Function  			: read message
 * InputParameter 		: voidevent
 * OutParameter			: NULL
 * Author 				: Dingjianming
 * Date					: 2016/08/17
 *
 * ==================================================*/
int SendMq::Amqp_First_Consume(const char* _queuename)
{
    try{
        amqp_basic_qos_ok_t *qos = nullptr;
        amqp_basic_consume_ok_t * consumeqos = NULL;
        qos = amqp_basic_qos(conn, channelid, 0, 1, 0);
        if(qos == nullptr)
        {
            LOG.error("C_Amqp_First_Consume Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }
        consumeqos = amqp_basic_consume(conn, channelid, amqp_cstring_bytes(_queuename),amqp_empty_bytes, 0, 0, 0,amqp_empty_table);
        if(consumeqos == nullptr)
        {
            LOG.error("C_Amqp_First_Consume Failed:%s,%d",__FILE__,__LINE__);
            return -1;
        }
    }catch(Poco::Exception& e)
    {
        LOG.fatal("Amqp_First_Consume:%s,%d",e.message(),__LINE__);
        return -1;
    }
    return 0;
}
/*====================================================
 * Name 	  			: C_Amqp_Basic_Consume
 * Function  			: read message
 * InputParameter 		: voidevent
 * OutParameter			: NULL
 * Author 				: Dingjianming
 * Date					: 2016/07/25
 * ==================================================*/
#if 0
std::string SendMq::Amqp_Basic_Consume(void)
{
    string returntemp;
    try
    {
        char tempstr[256]={0};
        char *tempData,*Head;
        MqReviceNode revicenode;
        amqp_frame_t frame;
        amqp_basic_deliver_t *d;
        amqp_basic_properties_t *p;
        uint64_t body_target;
        size_t body_received;
        timeval tv;
        tv.tv_sec = 0;
        tv.tv_usec = 500;
        int result;
        amqp_maybe_release_buffers(conn);
        result = amqp_simple_wait_frame_noblock(conn, &frame,&tv);//获取读取frame的编号
        //result = amqp_simple_wait_frame(conn, &frame);
        if(result < 0)
            return "NULL";

        revicenode.FrameType = frame.frame_type;
        revicenode.channel = frame.channel;
    #if 0
        cout << "Frame Type" << frame.frame_type << "channel:" << frame.channel << endl;
    #endif
        if(frame.frame_type != AMQP_FRAME_METHOD)
        {
            LOG.error("frame_type != AMQP_FRAME_METHOD:%s,%d", __FILE__,__LINE__);
            return "NULL";//继续
        }
        revicenode.method = amqp_method_name(frame.payload.method.id);
    #if 0
        cout << "Method:" << amqp_method_name(frame.payload.method.id);
    #endif
        if(frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
        {
            LOG.error("frame.payload.id != AMQP_BASIC_DELIVER_METHOD:%s,%d", __FILE__,__LINE__);
            return "NULL";
        }
        d = (amqp_basic_deliver_t *)frame.payload.method.decoded;
        revicenode.delivery = (unsigned)d->delivery_tag;
        sprintf(tempstr,"%.*s",(int)d->exchange.len,(char*)d->exchange.bytes);
        revicenode.exchange = tempstr;
        memset(tempstr,0,sizeof(tempstr));
        sprintf(tempstr,"%.*s",(int)d->routing_key.len,(char*)d->routing_key.bytes);
        revicenode.routkey = tempstr;
    #if 0
        cout << "Delivery:" << (unsigned)d->delivery_tag << "exchange:" <<(int)d->exchange.len<<"routkey:"<<(char*)d->exchange.bytes << (int)d->routing_key.len << (char*) d->routing_key.bytes<< endl;
    #endif
        result = amqp_simple_wait_frame_noblock(conn, &frame,&tv);
        //result = amqp_simple_wait_frame(conn, &frame);
        if(result < 0)
            return "NULL";
        if(frame.frame_type != AMQP_FRAME_HEADER)
            return "NULL";
        p = (amqp_basic_properties_t *)frame.payload.properties.decoded;
        if(p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG)
        {
            memset(tempstr,0,sizeof(tempstr));
            sprintf_s(tempstr,"%.*s",(int)p->content_type.len,(char*)p->content_type.bytes);
            revicenode.Content_Type = tempstr;
        }

        body_target = frame.payload.properties.body_size;
        body_received = 0;

        int sleep_seconds = 0;
        while(body_received < body_target)
        {
            result = amqp_simple_wait_frame_noblock(conn, &frame,&tv);
            //result = amqp_simple_wait_frame(conn, &frame);
            if(result < 0)
                return "NULL";
            if(frame.frame_type != AMQP_FRAME_BODY)
            {
                LOG.error("frame_type != AMQP_FRAME_BODY:%s,%d", __FILE__,__LINE__);
                return "NULL";
            }
            body_received += frame.payload.body_fragment.len;
            tempData = Head = (char*)malloc(1048576);
            if(body_received > 1048576)//1M
            {
                LOG.error("frame.payload.body_fragment.len > 1M:%s,%d", __FILE__,__LINE__);
                revicenode._data = "NULL";
                returntemp.append(revicenode._data);
                return returntemp;
            }
            int i;
            for(i = 0; i<frame.payload.body_fragment.len;i++)
            {
                *tempData = *((char*)frame.payload.body_fragment.bytes+i);
                tempData++;
#if 1
            //	cout << *((char*)frame.payload.body_fragment.bytes+i);
                if((*(char*)frame.payload.body_fragment.bytes+i))
                    sleep_seconds++;
#endif
            }
            *tempData = '\0';
        }
        if(strlen(Head) == 0)
        {
            revicenode._data = "NULL";
        }
        else
        {
            revicenode._data = Head;
            if(body_received != body_target)
            {
                LOG.error("body_received <= body_target:%s,%d", __FILE__,__LINE__);
                return "NULL";
            }
    #if 0
        Sleep(sleep_seconds);
    #endif
            amqp_basic_ack(conn, channelid, d->delivery_tag, 0);
        }
        free(Head);
        Head = tempData = NULL;
        returntemp.append(revicenode._data);


    }catch(Poco::Exception& e)
    {
        LOG.error("Amqp_Basic_Consume: %s,%d", e.message(),__LINE__);
        return "NULL";
    }
    return returntemp;
}
#endif
/*====================================================
 * Name 	  			: C_Amqp_Basic_Publish
 * Function  			: send message
 * InputParameter 	: voidevent
 * OutParameter		: NULL
 * Author 				: Dingjianming
 * Date					: 2016/07/22
 *
 * ==================================================*/
int SendMq::Amqp_Basic_Publish(const char* _exchange, const char* _routingkey,const char* messagebody)
{
    try{
        int ret;
        amqp_basic_properties_t props;
        props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
        props.content_type = amqp_cstring_bytes("json/plain");
        props.delivery_mode = 2;
        if (!MQFlag)
        {
            ret = amqp_basic_publish(conn, channelid,amqp_cstring_bytes(_exchange),amqp_cstring_bytes(_routingkey),0,0,&props,amqp_cstring_bytes(messagebody));
            if(ret < 0)
            {
                LOG.error("Publishing Falied:%s,%d", __FILE__,__LINE__);
                return -1;
            }
            _cException->die_on_error(ret,"Publishing");
        }
        else
        {
//            Amqp_Init_Mq();
//            ret = amqp_basic_publish(conn, channelid,amqp_cstring_bytes(_exchange),amqp_cstring_bytes(_routingkey),0,0,&props,amqp_cstring_bytes(messagebody));
//            if(ret < 0)
//            {
//                LOG.error("Publishing Falied:%s,%d", __FILE__,__LINE__);
//                return -1;
//            }
//            _cException->die_on_error(ret,"Publishing");
        }
    }
    catch(Poco::Exception& e)
    {
        LOG.error("Amqp_Basic_Publish: %s,%d", e.message(),__LINE__);
        return -1;
    }
    return 0;
}

